package com.shujia.rec.compute

import com.shujia.rec.common.Constants
import com.shujia.rec.entry.CaseClass
import com.shujia.rec.util.{KafkaUtil, LogUtil}
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.api.scala._
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.sink.{RichSinkFunction, SinkFunction}
import org.apache.hadoop.hbase.client._
import org.apache.hadoop.hbase.util.Bytes

object ItemCfHistory {
  def main(args: Array[String]): Unit = {

    /**
      *
      * 用户 --->  商品
      * 用户作为rowkey    商品作为列名 多次购买作为列值
      *
      * 商品 ---> 用户
      * 商品作为rowkey    用户作为列名 多次购买作为列值
      *
      */


    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)
    //获取kafka source
    val kafkaSource = KafkaUtil.getKafaSoure("ItemCfHistory", Constants.KAFKA_TOPIC)

    //读所有数据
    kafkaSource.setStartFromLatest()

    val logDS = env.addSource(kafkaSource)

    //过滤脏数据
    val filterDS = logDS.filter(log => LogUtil.verifyLog(log))

    //将每一行数据转换成实体类
    val entryDS = filterDS.map(log => LogUtil.toEntry(log))

    val actionDS = entryDS.filter(_.action == 3)


    //数据保存到hbase
    actionDS.addSink(sinkFunction = new RichSinkFunction[CaseClass.LogEntry] {

      var connection: HConnection = _
      var p: HTableInterface = _
      var u: HTableInterface = _

      override def open(parameters: Configuration): Unit = {
        //换件hbase连接
        val configuration = new org.apache.hadoop.conf.Configuration()
        configuration.set("hbase.zookeeper.quorum", Constants.HBASE_ZOOKEEPER_CONNECT)
        connection = HConnectionManager.createConnection(configuration)

        /**
          * create 'p_history','info'
          *
          * create 'u_history','info'
          */

        p = connection.getTable("rec:p_history")
        u = connection.getTable("rec:u_history")

      }

      override def close(): Unit = {
        if (connection != null) {
          connection.close()
        }
      }

      override def invoke(value: CaseClass.LogEntry, context: SinkFunction.Context[_]): Unit = {
        inc(p, String.valueOf(value.proId), String.valueOf(value.userId))
        inc(u, String.valueOf(value.userId), String.valueOf(value.proId))

        //将商品被购买的历史存到redis的位图中  共后续计算同现次数
        //jedis.setbit("p_history:" + value.proId, value.userId, "1")

      }

      //多一个rowkey中一个列进行递增
      def inc(table: HTableInterface, rowkey: String, colume: String): Unit = {
        val pget = new Get(rowkey.getBytes())
        //取出一列
        pget.addColumn("info".getBytes(), colume.getBytes)
        val prs = table.get(pget)
        val pc = Bytes.toString(prs.getValue("info".getBytes(), colume.getBytes))
        var pnum = 1
        if (pc != null) {
          pnum = pnum + Integer.parseInt(pc)
        }
        val put = new Put(rowkey.getBytes())
        put.add("info".getBytes(), colume.getBytes, String.valueOf(pnum).getBytes())

        table.put(put)
      }
    })

    env.execute("ItemCfHistory")
  }
}
